package com.chis.flink;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import java.util.List;

public class MySqlDsTest extends RichSourceFunction<FlinkTestBean>  {

    private ApplicationContext applicationContext;
    private IMysqlTest mysqlTest;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        if(applicationContext == null){
            applicationContext = new ClassPathXmlApplicationContext(new String[] { "applicationContext.xml" });
            mysqlTest = (IMysqlTest) applicationContext.getBean("qqqqq");
        }

    }

    @Override
    public void run(SourceContext<FlinkTestBean> sourceContext) throws Exception {
        List<FlinkTestBean> list = mysqlTest.findTest();
        list.parallelStream().forEach(s -> sourceContext.collect(s));

    }

    @Override
    public void cancel() {

    }

//    @Override
//    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
//        this.applicationContext = applicationContext;
//    }
}
